package org.shj.spark.udf

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.StructField

/**
 * UDF: User Defined Function. 函数的输入是一条具体的数据记录，实现上讲就是普通的scala函数
 * UDAF: User Defined Aggregation Function。 函数本身作用于数据集合，能够在聚合操作的基础上进行自定义的操作
 * 
 * 实质上讲，例如说 UDF 会被 Spark SQL 中的 Catalyst 封装成 Expression，最终会通过eval方法来计算输入
 * 的数据Row(此处的 Row 和 DataFrame 中的 Row 没有任何关系)
 */
object SparkSqlUDF {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("SparkSqlUDF").master("local").getOrCreate();
    spark.sparkContext.setLogLevel("WARN")
    
    import spark.implicits._
    
    val datas = Array("YangMi", "ZhaoLiYing ", " ZhuYin")
    spark.udf.register("computeLength", (input: String) => input.trim().length())
    
    val rowRDD = spark.sparkContext.parallelize(datas).map(item => Row(item))
    val schema = StructType(Array(StructField("name", StringType)))
    
    val df = spark.createDataFrame(rowRDD, schema)
    df.createOrReplaceTempView("tmpTable")
    spark.sql("select name, computeLength(name) len from tmpTable").show()
    
    spark.close()
  }
}